package com.cloud.configure;

import cn.hutool.core.util.RandomUtil;
import com.cloud.message.AbstractQTMessage;
import com.cloud.mqtt.QTConsumer;
import com.cloud.mqtt.QTMode;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Map;

/**
 * Mqtt的配置类
 * 服务端 EMQX服务: https://www.emqx.io/zh/downloads
 *
 * @author yzj
 */
@Configuration
public class MqttConfiguration implements ApplicationContextAware {

    private static Logger logger = LoggerFactory.getLogger(MqttConfiguration.class);

    @Resource
    private MqttProperties mqttProperties;

    private ConfigurableApplicationContext applicationContext;

    private MqttClient mqttProducer;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }

    /**
     * 创建Mqtt连接器选项
     */
    @Bean
    public MqttConnectOptions getOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        // 设置连接的用户名
        if (StringUtils.isNotEmpty(mqttProperties.getUserName())) {
            // 将用户名去掉前后空格
            options.setUserName(mqttProperties.getUserName());
        }
        // 设置连接的密码
        options.setPassword(mqttProperties.getPassword().toCharArray());
        // 转化连接的url地址
        String[] urls = {mqttProperties.getHostUrl()};
        // 设置连接的地址
        options.setServerURIs(urls);
        // 设置超时时间单位为秒
        options.setConnectionTimeout(mqttProperties.getCompletionTimeout());
        /**
         * 设置会话心跳时间单位为秒,服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
         * 但这个方法并没有重连的机制
         */
        options.setKeepAliveInterval(mqttProperties.getKeepAliveInterval());
        // 设置“遗嘱”消息的话题,若客户端与服务器之间的连接意外中断,服务器将发布客户端的“遗嘱”消息
        // 设置超时时间
        options.setConnectionTimeout(mqttProperties.getConnectionTimeout());
        options.setCleanSession(true);
        // 设置断开后重新连接
        options.setAutomaticReconnect(true);
        return options;
    }

    /**
     * 发布者对象
     * 这里订阅者和发布者的MqttClient对象分别命名是为了让发布者和订阅者分开，
     * 如果订阅者和发布者都用一个MqttClient链接对象,则会出现两方都订阅了某个主题后，
     * 谁发送了消息,都会自己接收到自己发的消息,所以分开写,里面主要就是回调类的设置setCallback
     */
    @Bean
    public MqttClient getMqttProducer() {
        try {
            String clientCode = RandomUtil.randomNumbers(6);
            String clientId = mqttProperties.getClientKey() + "_" + clientCode;
            // 判断客户端链接状态
            mqttProducer = new MqttClient(mqttProperties.getHostUrl(), clientId, new MemoryPersistence());
            if (!mqttProducer.isConnected()) {
                MqttConnectOptions options = this.getOptions();
                mqttProducer.connect(options);
            }
            logger.info("MQTT生产者:{},启动成功", clientId);
        } catch (Exception e) {
            logger.error("MQTT生产者,启动失败", e);
        }
        return mqttProducer;
    }

    /**
     * 订阅者对象
     *
     * @return
     * @throws MqttException
     */
    @PostConstruct
    public void qtConsumer() {
        Map<String, Object> consumerBeans = applicationContext.getBeansWithAnnotation(QTConsumer.class);
        if (ObjectUtils.isNotEmpty(consumerBeans)) {
            for (Map.Entry<String, Object> entry : consumerBeans.entrySet()) {
                loadQTConsumer(entry.getKey(), entry.getValue());
            }
        }
    }

    private void loadQTConsumer(String beanName, Object bean) {
        try {
            QTConsumer QTConsumer = applicationContext.findAnnotationOnBean(beanName, QTConsumer.class);
            Assert.notNull(QTConsumer.topic(), "MQTT必须定义订阅消息的主题");
            Assert.notNull(QTConsumer.qos(), "MQTT必须定义订阅消息的级别");
            if (!AbstractQTMessage.class.isAssignableFrom(bean.getClass())) {
                throw new RuntimeException(bean.getClass().getName() + "-consumer未实现Consumer抽象类");
            }
            String subTopic = null;
            if (QTMode.NONE.equals(QTConsumer.mode())) {
                subTopic = QTConsumer.topic();
            } else if (QTMode.SHARE.equals(QTConsumer.mode())) {
                subTopic = "$share/" + QTConsumer.group() + "/" + QTConsumer.topic();
            } else if (QTMode.QUEUE.equals(QTConsumer.mode())) {
                subTopic = "$share/$queue/" + QTConsumer.topic();
            } else {
                subTopic = QTConsumer.topic();
            }
            AbstractQTMessage qtPushConsumer = (AbstractQTMessage) bean;
            qtPushConsumer.setClientConnect(mqttProperties.getHostUrl(),
                    mqttProperties.getClientKey(), subTopic, QTConsumer.qos());
            MqttClient mqttConsumer = qtPushConsumer.getMqttClient(this.getOptions(), bean);
            // 设置MqttClient订阅端回调函数
            mqttConsumer.setCallback(qtPushConsumer);
            mqttConsumer.subscribe(subTopic, QTConsumer.qos(), (String topic, MqttMessage message) ->
                    qtPushConsumer.messageArrived(topic, message));
            logger.info("MQTT订阅者:{}开始", mqttConsumer.getClientId());
        } catch (Exception e) {
            logger.error("MQTT订阅主题:{},失败", e);
        }
    }

}
